# -*- coding: utf-8 -*-

"""
kg json to neo4j csv job
"""

from __future__ import unicode_literals
from __future__ import absolute_import
from __future__ import division

from kgpipeline.job import KgJob
from kgpipeline.jsonutil import entity_json_decode
from kgpipeline.tools.json2csv import json_rdd_to_csv_df
from kgpipeline.sparkutil import current_session


class KgNeo4jCSVJob(KgJob):
    """job for convert json to csv for neo4j"""
    def process(self, inputs):
        session = current_session()
        res = []
        for df, _ in inputs:
            if not df or df.rdd.isEmpty():
                res.append(None)
            else:
                res.append(json_rdd_to_csv_df(session, df.rdd.map(lambda x: entity_json_decode(x["value"]))))

        return res


KgJob.register("neo4jcsv", KgNeo4jCSVJob)
